package com.orion.service;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author Administrator
 * @date 2021/12/2
 */
@Component
public class MessageConsumerService {

    /**
     * 设置consumer的auto_offset_set为earliest，
     * b项目设置latest，测试latest不会消费已存在的offset只会消息最新生产的消息，而earliest则会消费
     * <p>
     * <p>
     * 和b项目监听同一个topic，而且是同一个group，即等于集群，默认concurrcy是3，先后启动ab，可见重平衡的情况
     *
     * @param records
     * @param ack
     */
    @KafkaListener(id = "xixi", groupId = "sadan", topics = {"llc2021","llc2022"}, concurrency = "2")
    public void singleConsume01(List<ConsumerRecord<String, Object>> records, Acknowledgment ack) {
        System.out.println("--------- 接收并处理消息 ---------");
        for (ConsumerRecord<String, Object> record : records) {
            System.out.println(Thread.currentThread() + "-----消费者收到消息:" + record + "; ack:" + ack);
        }
        ack.acknowledge();
    }
}
